package com.syy.outputformat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class LogRecordWriter extends RecordWriter<LongWritable, Text> {
    private FSDataOutputStream fs1;
    private FSDataOutputStream fs2;

    public LogRecordWriter(TaskAttemptContext job) throws IOException {
        // 拿到全局配置
        Configuration configuration = job.getConfiguration();
        // 创建FileSystem文件对象
        FileSystem fs = FileSystem.get(configuration);
        // 创建输出流
        fs1 = fs.create(new Path(configuration.get(FileOutputFormat.OUTDIR) + "\\itwwy.log"));
        fs2 = fs.create(new Path(configuration.get(FileOutputFormat.OUTDIR) + "\\other.log"));
    }

    @Override
    public void write(LongWritable key, Text value) throws IOException, InterruptedException {
        String log = value.toString()+"\n";
        if (log.contains("itwwy")) {
            // 写入到itwwy.log文件
            fs1.write(log.getBytes());
        } else {
            // 写入到other.log文件
            fs2.write(log.getBytes());
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        // 关闭
        IOUtils.closeStream(fs1);
        IOUtils.closeStream(fs2);
    }
}
